package com.lry;

import com.lry.app.ExampleStateMachine;
import com.lry.config.BrokerConfig;
import com.lry.config.FlushDiskType;
import com.lry.config.MessageStoreConfig;
import com.lry.core.RaftNode;
import com.lry.core.RaftOptions;
import com.lry.core.Server;
import com.lry.raft.MqRaftServiceImpl;
import com.lry.raft.MqStateMachine;
import com.lry.rpc.MqRpcServiceImpl;
import com.lry.server.RpcServer;
import com.lry.service.impl.RaftConsensusServiceImpl;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

public class RaftBrokerStartup {

    public static void main(String[] args) throws IOException {

        List<BrokerConfig> brokerConfigs = new ArrayList<>();
        brokerConfigs.add(new BrokerConfig("broker1",1,8081));
        brokerConfigs.add(new BrokerConfig("broker1",2,8082));
        brokerConfigs.add(new BrokerConfig("broker1",3,8083));

        List<Server> servers = new ArrayList<>();
        for (BrokerConfig brokerConfig : brokerConfigs) {
            servers.add(new Server(brokerConfig.getBrokerId(),brokerConfig.getHost(),brokerConfig.getPort()));
        }


        for (int i=0;i<servers.size();i++) {
            int finalI = i;
            CompletableFuture.runAsync(()->{
                BrokerConfig brokerConfig = brokerConfigs.get(finalI);
                Server server = servers.get(finalI);
                BrokerController brokerController = null;
                try {
                    brokerController = new BrokerController(brokerConfig,buildMessageStoreConfig(finalI));
                } catch (Exception e) {
                    e.printStackTrace();
                }

                RaftOptions raftOptions = new RaftOptions();
                MqStateMachine stateMachine = new MqStateMachine(brokerController);
                RaftNode raftNode = new RaftNode(raftOptions,server,servers,stateMachine);

                RpcServer rpcServer = new RpcServer(brokerConfig.getHost(),brokerConfig.getPort());

                rpcServer.registerService(new MqRpcServiceImpl(brokerController));//注册client访问broker的api
                rpcServer.registerService(new MqRaftServiceImpl(raftNode));//注册client通过raft访问broker的api
                rpcServer.registerService(new RaftConsensusServiceImpl(raftNode));//注册raft节点间的api，请求投票，心跳，追加日志

                rpcServer.start();
                raftNode.start();
            });
        }

        System.in.read();
    }

    //测试使用
    private static MessageStoreConfig buildMessageStoreConfig(int index)   {
        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
        messageStoreConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "store"+index);
        messageStoreConfig.setStorePathCommitLog(System.getProperty("user.home") + File.separator + "store"+index+File.separator + "commitlog");
        messageStoreConfig.setMappedFileSizeCommitLog(1024 * 1024 * 10);
        messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 1024 * 10);
        messageStoreConfig.setMaxHashSlotNum(10000);
        messageStoreConfig.setMaxIndexNum(100 * 100);
        messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
        messageStoreConfig.setFlushIntervalConsumeQueue(1);

        return messageStoreConfig;
    }
}
